package com.mimo.logic.message.service.impl;

import java.time.Duration;
import java.util.stream.Stream;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.mimo.common.configuration.mrlock.annotation.DistributedRedisLock;
import com.mimo.logic.message.constants.MessageKeys;
import com.mimo.logic.message.service.IMessageScheduleService;
import com.mimo.logic.message.service.IMessageDispatcher;

@Service
public class MessageClearServiceImpl implements IMessageScheduleService {

  private static final Logger log = LoggerFactory.getLogger(MessageClearServiceImpl.class);

  @Autowired
  private IMessageDispatcher messageDispatcher;

  @Override
  @DistributedRedisLock(key = MessageKeys.LOGIC_MSG_SCHEDULE_CLEAR_KEY, expired = 60, waited = 60)
  public void clear(Duration expired, long count) {
    long end = System.currentTimeMillis() - expired.toMillis();
    Stream.iterate(0, e -> e + 1).limit(IMessageDispatcher.MAX_SLOT)
        .flatMap(slot -> messageDispatcher.getP2PMsgByBucketSlotAndExpireBetween(slot, -1, end, count).stream())
        .forEach(mid -> {
          log.warn("message[{}] 过期", mid);
          messageDispatcher.remove(mid);
        });
  }

}
